from collections import defaultdict, deque
import datetime
import kafka
import logging
import math
import os
import pytz
import random
import typing
import ujson as json

import coin.proto.coin_market_enums_pb2 as cme_pb2
from coin.proto.coin_meta_service_pb2 import QueryTransferRequestProto
from coin.proto.coin_request_pb2 import TransferRequestProto
from coin.proto.coin_strategy_summary_pb2 import StrategySummary
from coin.base.datetime_util import (to_datetime, to_timestamp_int)
from coin.base.config_util import load_clean_json
from coin.base.param_util import to_list
from coin.util.queue.constants import TopicType
from coin.util.queue.config import KafkaConfig
from coin.util.queue.tools.kafka_topic import parse_kafka_topic
from xunkemgmt_client.client.api_client import XunkemgmtClient
from xunkemgmt_client.client.util.query_util import query_strategies


def _filter_strat_topic(topics, strats, topic_type: TopicType):
  topic_type_candiate = topic_type if topic_type is not None else TopicType.STRAT
  filtered = []
  not_found = []
  strat_topic_map = defaultdict(list)
  for topic in topics:
    try:
      topic_type, topic_info = parse_kafka_topic(topic)
    except:
      logging.info('Fail to parse topic: %s' % topic)
      continue
    if topic_type != topic_type_candiate:
      continue
    strat_topic_map[topic_info.strategy_name].append(topic)
  for strat in strats:
    if strat in strat_topic_map:
      filtered += strat_topic_map[strat]
    else:
      not_found.append(strat)
  if len(not_found) > 0:
    logging.info('Strategies not found in topics: %s' % not_found)
  return filtered


def _filter_account_topic(topics, accounts):
  filtered = []
  not_found = []
  account_topic_map = defaultdict(list)
  acct_key_func = lambda x: (x.market_type, x.exchange, x.owner)
  for topic in topics:
    try:
      topic_type, topic_info = parse_kafka_topic(topic)
    except:
      logging.info('Fail to parse topic: %s' % topic)
      continue
    if topic_type != TopicType.ACCOUNT:
      continue
    account_topic_map[acct_key_func(topic_info)].append(topic)
  for account in accounts:
    acct_key = acct_key_func(account)
    if acct_key in account_topic_map:
      filtered += account_topic_map[acct_key]
    else:
      not_found.append(acct_key)
  if len(not_found) > 0:
    logging.info('Accounts not found in topics: %s' % not_found)
  return filtered


def _filter_ticker_topic(topics, sub_reqs):
  filtered = []
  not_found = []
  sub_topic_map = defaultdict(list)
  sub_key_func = lambda x: (x.market_type, x.exchange, x.api_version)
  for topic in topics:
    try:
      topic_type, topic_info = parse_kafka_topic(topic)
    except:
      logging.info('Fail to parse topic: %s' % topic)
      continue
    if topic_type != TopicType.TICKER:
      continue
    sub_topic_map[sub_key_func(topic_info)].append(topic)
  for sub_req in sub_reqs:
    sub_key = sub_key_func(sub_req)
    if sub_key in sub_topic_map:
      filtered += sub_topic_map[sub_key]
    else:
      not_found.append(sub_key)
  if len(not_found) > 0:
    logging.info('Subscriptions not found in topics: %s' % not_found)
  return filtered


def get_active_strategy(*, business_unit=None, update_timestamp_threshold=None):
  strategies = []
  biz_units = [] if business_unit is None else to_list(business_unit)
  for elem in query_strategies(business_units=biz_units, as_proto=True):
    if elem.active:
      strategies.append(elem.strategy_name)
    else:
      if update_timestamp_threshold is not None and \
        elem.HasField('update_timestamp') and \
        elem.update_timestamp > update_timestamp_threshold:
        strategies.append(elem.strategy_name)
  return list(set(strategies))


def get_topics(strat_list: typing.Optional[typing.List[str]],
               kafka_config: KafkaConfig,
               *,
               business_units: typing.Optional[typing.List[str]]=None,
               topic_type: typing.Optional[TopicType]=None
    ) -> typing.List[str]:
  if strat_list:
    topics = get_topic_from_strat(
        strat_list, kafka_config, topic_type=topic_type)
  else:
    threshold = to_timestamp_int(
        datetime.datetime.utcnow().replace(tzinfo=pytz.UTC) - datetime.timedelta(days=1))
    active_strats = get_active_strategy(
        business_unit=business_units, update_timestamp_threshold=threshold)
    topics = get_topic_from_strat(active_strats , kafka_config, topic_type=topic_type)
  return topics


def get_topic_from_strat(
    strat_list, kafka_config, *,
    topic_type=None, business_unit=None, strategy_group=None, active=None):
  filtered = query_strategies(active=active, as_proto=True)
  if business_unit is not None:
    filtered = [elem for elem in filtered if elem.business_unit == business_unit]
  if strategy_group is not None:
    filtered = [elem for elem in filtered if elem.strategy_group == strategy_group]
  all_strats =  [elem.strategy_name for elem in filtered]
  if strat_list is not None:
    strat_list = list(set(strat_list).intersection(set(all_strats)))
  else:
    strat_list = all_strats
  all_topics = kafka.KafkaConsumer(bootstrap_servers=kafka_config.kafka_servers).topics()
  # sample topic name: STRAT_hive-29_unhedged_mm_okex_pilot
  topics = _filter_strat_topic(all_topics, strat_list, topic_type)
  return topics


def get_topic_from_config(
    strategy_config_filename, kafka_config_filename, *,
    topic_type=None, strategy_group=None):
  strats = get_strat_from_json_config(strategy_config_filename)
  kafka_config = KafkaConfig.from_cmd_config(kafka_config_filename)
  topics = get_topic_from_strat(
      strats, kafka_config, topic_type=topic_type, strategy_group=strategy_group)
  return topics


def get_strat_from_json_config(config_path):
  assert config_path and os.path.isfile(config_path), \
      'Config file not existed: %s' % config_path
  config = load_clean_json(config_path)
  strats = [entry['strategy_name'] if isinstance(entry, dict) else entry
            for entry in config]
  return list(set(strats))


def select_strategy_summary(summary_list, strategy_group=None, business_unit=None):
  result_list = summary_list
  if strategy_group is not None:
    strategy_group_list = [elem.strip() for elem in strategy_group.split(',')]
    result_list = [
        summary for summary in result_list if isinstance(summary, StrategySummary)
        and summary.strategy.strategy_group in strategy_group_list
    ]
  if business_unit is not None:
    business_unit_list = [elem.strip() for elem in business_unit.split(',')]
    result_list = [
        summary for summary in result_list if isinstance(summary, StrategySummary)
        and summary.strategy.business_unit in business_unit_list
    ]
  return result_list


def separate_into_group(topics: typing.List[str], max_group: int
    ) -> typing.List[typing.List[str]]:
  assert isinstance(topics, list) and len(topics) > 0
  num_per_group = math.ceil(len(topics) / max_group)
  random.shuffle(topics)
  groups = []
  group = []
  while len(topics) > 0:
    group.append(topics.pop())
    if len(group) == num_per_group:
      groups.append(group)
      group = []
  else:
    if len(group) > 0:
      groups.append(group)
  return groups


def get_pnl_adjust_info(pnl_adj_file, start_date, end_date):
  pnl_adj = load_pnl_adjust_config(pnl_adj_file)
  pnl_adj_from_transfer = load_pnl_adjust_config_from_transfer(
      QueryTransferRequestProto(
          transfer_status=TransferRequestProto.TRANSFER_CONFIRMED,
          start_completed_timestamp=to_timestamp_int(start_date),
          end_completed_timestamp=to_timestamp_int(end_date)))
  pnl_adj = {key: value for key, value in pnl_adj.items() if
             start_date <= datetime.datetime.strptime(key, '%Y%m%d').date() < end_date}
  total_pnl_adj = {}
  keys = set(pnl_adj.keys()).union(set(pnl_adj_from_transfer.keys()))
  for key in keys:
    total_pnl_adj[key] = pnl_adj.get(key, []) + pnl_adj_from_transfer.get(key, [])
  return total_pnl_adj


def load_pnl_adjust_config(pnl_adj_file):
  pnl_adj = {}
  if pnl_adj_file is not None:
    assert os.path.isfile(pnl_adj_file), \
        'pnl adj file does not exist %s' % pnl_adj_file
    with open(pnl_adj_file, 'r') as open_file:
      pnl_adj = json.load(open_file)
  return pnl_adj


def load_pnl_adjust_config_from_transfer(request):
  with XunkemgmtClient() as client:
    rsp = client.query_transfer_requests(request)
  data = defaultdict(list)
  for elem in rsp:
    if (elem.status == TransferRequestProto.TRANSFER_CONFIRMED and
        elem.product_type == cme_pb2.CURRENCY and
        elem.pnl_adj_needed):
      assert elem.HasField('completed_timestamp') and \
             elem.HasField('symbol') and \
             elem.HasField('qty') and \
             elem.HasField('type') and \
             elem.strategy.HasField('strategy_name'), elem
      td_str = to_datetime(elem.completed_timestamp).date().strftime('%Y%m%d')
      if elem.type == TransferRequestProto.DEPOSIT:
        sign = -1
      elif elem.type == TransferRequestProto.WITHDRAW:
        sign = 1
      else:
        raise ValueError(elem.type)
      adj_ccy = elem.symbol
      adj_qty = elem.qty
      if elem.HasField('pnl_adj_symbol') and elem.HasField('pnl_adj_qty'):
        adj_ccy = elem.pnl_adj_symbol
        adj_qty = elem.pnl_adj_qty
      data[td_str].append({
          'strategy_name': elem.strategy.strategy_name,
          'pnl_adj': abs(adj_qty) * sign,
          'accounting_currency': adj_ccy,
      })
  return data


def filter_empty_summary(summaries):
  result = []
  for summary in summaries:
    assert isinstance(summary, StrategySummary)
    if len(summary.stat_per_account) > 0 or len(summary.stat_per_quote) > 0:
      result.append(summary)
  return result
